package com.atguigu.pro

import java.text.SimpleDateFormat
import java.util

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
 * @description: 热门页面统计案例,所有窗口触发都会基于新的wm,watermark在51-1的时候50就输出一个,由于有延时数据那么则会聚合计算但是窗口并会不关闭,
 *               定时器的触发也是基于wm,只有wm更新了才向下游传递,同一个数据会在不同的窗口中输出结果
 * @time: 2021/3/30 14:17
 * @author: baojinlong
 * */
// 定义输入数据样例类
case class ApacheLogEvent(ip: String, userId: String, timestamp: Long, method: String, url: String)

// 先做开窗的聚合操作,得到聚合结果后再按照每个窗口排列,定义一个窗口聚合结果样例类
case class PageViewCount(url: String, windowEnd: Long, count: Long)

object HotPagesNetwork {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 读取数据,转换成样例类并提取时间戳和watermark
    //val inputStream: DataStream[String] = env.readTextFile("c:/apache.log")
    val inputStream: DataStream[String] = env.socketTextStream("localhost", 8888)
    val dataStream: DataStream[ApacheLogEvent] = inputStream.map(
      data => {
        val arr: Array[String] = data.split(" ")
        // 对事件时间进行转换,得到时间戳
        val simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
        val ts: Long = simpleDateFormat.parse(arr(3)).getTime
        ApacheLogEvent(arr(0), arr(1), ts, arr(5), arr(6))
      }
      // 这是乱序数据
    ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent](Time.seconds(1)) {
      override def extractTimestamp(t: ApacheLogEvent): Long = t.timestamp
    })


    // 进行开窗聚合,以及排序输出
    val aggStream: DataStream[PageViewCount] = dataStream
      .filter(_.method.equals("GET"))
      .keyBy(_.url)
      // 最近十分钟,每隔5s统计一次
      .timeWindow(Time.minutes(10), Time.seconds(5))
      // 在09:00:01的时候窗口还别关,这个一分钟的时间来的数据还都在做聚合统计,到09:01的时候就关闭窗口
      .allowedLateness(Time.seconds(1))
      // 三重保证->侧输出流
      .sideOutputLateData(new OutputTag[ApacheLogEvent]("late"))
      .aggregate(new PageCountAgg, new PageViewCountWindowResult)

    // 根据聚合结果流进行排序输出
    val resultStream: DataStream[String] = aggStream
      .keyBy(_.windowEnd)
      .process(new TopNHotPages(3))

    dataStream.print("data")
    aggStream.print("agg")
    aggStream.getSideOutput(new OutputTag[ApacheLogEvent]("late")).print("late")
    // 打印输出
    resultStream.print("resultStream")
    env.execute("hotPageTest")
  }

}

class PageCountAgg extends AggregateFunction[ApacheLogEvent, Long, Long] {
  override def createAccumulator(): Long = 0

  override def add(in: ApacheLogEvent, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1
}

// IN, OUT, KEY, W 输入为预聚合结果输出 输出为当前定义的样例类类型
class PageViewCountWindowResult extends WindowFunction[Long, PageViewCount, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[PageViewCount]): Unit = {
    // 包装成输出对象,窗口关闭的时候会计算一次
    out.collect(PageViewCount(key, window.getEnd, input.iterator.next))
  }
}

// key为windowEnd,输入为样例类,输出为String类型
class TopNHotPages(n: Int) extends KeyedProcessFunction[Long, PageViewCount, String] {
  lazy val pageViewCountListState: ListState[PageViewCount] = getRuntimeContext.getListState(new ListStateDescriptor("pageViewCount-list", classOf[PageViewCount]))

  override def processElement(i: PageViewCount, context: KeyedProcessFunction[Long, PageViewCount, String]#Context, collector: Collector[String]): Unit = {
    // 添加到集合中
    pageViewCountListState.add(i)
    // 注册定时器
    context.timerService.registerEventTimeTimer(i.windowEnd + 1)
  }

  // 定时器触发后实现逻辑
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, PageViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
    val allPageViewCounts: ListBuffer[PageViewCount] = ListBuffer()
    val iterator: util.Iterator[PageViewCount] = pageViewCountListState.get.iterator
    while (iterator.hasNext) {
      allPageViewCounts += iterator.next
    }
    // 提前清空状态
    pageViewCountListState.clear()

    // 按照访问量排名并输出top n
    val sortedPageViewCounts: ListBuffer[PageViewCount] = allPageViewCounts.sortWith(_.count > _.count).take(n)
    // 将排名信息格式化String,便于打印输出可视化输出
    val result = new StringBuilder
    // 遍历结果列表中的每一个PageViewCount数据,输出到一行
    for (i <- sortedPageViewCounts.indices) {
      // 获取每一个内容
      val currentItemViewCount: PageViewCount = sortedPageViewCounts(i)
      result.append("NO").append("i+1").append(":\t")
        .append("页面URL=").append(currentItemViewCount.url).append("\t")
        .append("热门度=").append(currentItemViewCount.count).append("\n")
    }
    result.append("\n==============\n\n")
    Thread.sleep(1000)
    out.collect(result.toString)
  }
}